---
title: "Agent Stack"
icon: "server"
---

[Agent Stack](https://beeai.dev/) is an open platform to help you discover, run, and compose AI agents from any framework.
This tutorial demonstrates how to consume agents from the Agent Stack and expose agents built in BeeAI Framework to the Agent Stack.

**Prerequisites**

- **[Agent Stack](https://beeai.dev/)** installed and running locally
- **BeeAI Framework** installed with `pip install beeai-framework[agentstack]`

---


## Consuming from the platform (client)

The `AgentStackAgent` class allows you to connect to any agent hosted on the Agent Stack. This means that you can interact with agents built from any framework!

Use `AgentStackAgent` when:
- You're connecting specifically to the Agent Stack services.
- You want forward compatibility for the Agent Stack, no matter which protocol it is based on.


Here's a simple example that uses the built-in `chat` agent:


<CodeGroup>

	{/* <!-- embedme python/examples/agents/providers/agent_stack.py --> */}

	```py Python [expandable]
	import asyncio
	import sys
	import traceback
	
	from beeai_framework.adapters.agentstack.agents import AgentStackAgent
	from beeai_framework.errors import FrameworkError
	from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory
	from examples.helpers.io import ConsoleReader
	
	
	async def main() -> None:
	    reader = ConsoleReader()
	
	    agents = await AgentStackAgent.from_agent_stack(url="http://127.0.0.1:8333", memory=UnconstrainedMemory())
	    if len(agents) > 1:
	        reader.write("Prompt: ", "Select one of the available agents:\n")
	        while True:
	            for index, agent in enumerate(agents):
	                reader.write("AgentStack: ", f"{index}) {agent.name} - {agent.meta.description}")
	
	            agents_index = reader.ask_single_question("Write agent's number: ")
	            try:
	                agent = agents[int(agents_index)]
	                if agent:
	                    break
	
	            except (ValueError, IndexError):
	                reader.write(
	                    "AgentStack (error) : ",
	                    f"Invalid selection: `{agents_index}`. Please enter a valid agent number.\n",
	                )
	    elif len(agents) == 1:
	        agent = agents[0]
	    else:
	        reader.write("AgentStack (error) : ", "No agent registered within the agent stack.\n")
	        exit(0)
	
	    reader.write("AgentStack: ", f"Selected {agent.name}:\n")
	    for prompt in reader:
	        # Run the agent and observe events
	        response = await agent.run(prompt).on(
	            "update",
	            lambda data, event: (reader.write(f"{agent.name} 🤖 (debug) : ", data)),
	        )
	
	        reader.write(f"{agent.name} Agent 🤖 : ", response.last_message.text)
	
	
	if __name__ == "__main__":
	    try:
	        asyncio.run(main())
	    except FrameworkError as e:
	        traceback.print_exc()
	        sys.exit(e.explain())
	
	```

	{/* <!-- embedme typescript/examples/agents/providers/agent_stack.ts --> */}
	```ts TypeScript [expandable]
	import "dotenv/config.js";
	import { AgentStackAgent } from "beeai-framework/adapters/agentstack/agents/agent";
	import { createConsoleReader } from "examples/helpers/io.js";
	import { FrameworkError } from "beeai-framework/errors";
	import { TokenMemory } from "beeai-framework/memory/tokenMemory";
	
	////////////////////////////////////////////////////////
	///   Supports only BeeAI platform version v0.2.xx   ///
	////////////////////////////////////////////////////////
	
	const agentName = "chat";
	
	const instance = new AgentStackAgent({
	  url: "http://127.0.0.1:8333/api/v1/acp",
	  agentName,
	  memory: new TokenMemory(),
	});
	
	const reader = createConsoleReader();
	
	try {
	  for await (const { prompt } of reader) {
	    const result = await instance.run({ input: prompt }).observe((emitter) => {
	      emitter.on("update", (data) => {
	        reader.write(`Agent (received progress) 🤖 : `, JSON.stringify(data.value, null, 2));
	      });
	      emitter.on("error", (data) => {
	        reader.write(`Agent (error) 🤖 : `, data.message);
	      });
	    });
	
	    reader.write(`Agent (${agentName}) 🤖 : `, result.result.text);
	  }
	} catch (error) {
	  reader.write("Agent (error)  🤖", FrameworkError.ensure(error).dump());
	}
	
	```
</CodeGroup>


**Usage in Workflow**

You can compose multiple Agent Stack agents into advanced workflows using the BeeAI framework's workflow capabilities. This example demonstrates a research and content creation pipeline:

In this example, the `GPT Researcher` agent researches a topic, and the `Podcast creator` takes the research report and produces a podcast transcript.

You can adjust or expand this pattern to orchestrate more complex multi-agent workflows.

{/* <!-- embedme python/examples/workflows/remote.py --> */}

```py Python [expandable]
import asyncio
import sys
import traceback

from pydantic import BaseModel

from beeai_framework.adapters.agentstack import AgentStackAgent
from beeai_framework.errors import FrameworkError
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory
from beeai_framework.workflows import Workflow
from examples.helpers.io import ConsoleReader


async def main() -> None:
    reader = ConsoleReader()

    class State(BaseModel):
        topic: str
        research: str | None = None
        output: str | None = None

    agents = await AgentStackAgent.from_agent_stack(url="http://127.0.0.1:8333", memory=UnconstrainedMemory())

    async def research(state: State) -> None:
        # Run the agent and observe events
        try:
            research_agent = next(agent for agent in agents if agent.name == "GPT Researcher")
        except StopIteration:
            raise ValueError("Agent 'GPT Researcher' not found") from None
        response = await research_agent.run(state.topic).on(
            "update",
            lambda data, _: (reader.write("Agent 🤖 (debug) : ", data)),
        )
        state.research = response.last_message.text

    async def podcast(state: State) -> None:
        # Run the agent and observe events
        try:
            podcast_agent = next(agent for agent in agents if agent.name == "Podcast creator")
        except StopIteration:
            raise ValueError("Agent 'Podcast creator' not found") from None
        response = await podcast_agent.run(state.research or "").on(
            "update",
            lambda data, _: (reader.write("Agent 🤖 (debug) : ", data)),
        )
        state.output = response.last_message.text

    # Define the structure of the workflow graph
    workflow = Workflow(State)
    workflow.add_step("research", research)
    workflow.add_step("podcast", podcast)

    # Execute the workflow
    result = await workflow.run(State(topic="Connemara"))

    print("\n*********************")
    print("Topic: ", result.state.topic)
    print("Research: ", result.state.research)
    print("Output: ", result.state.output)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

```


## Exposing to the platform (server)

The `AgentStackServer` class exposes an agent or any other runnable (tool/chat model, ...) to the **Agent Stack**.
It gets automatically registered to the platform and allows you to access and use the agents directly in the platform.

Key Features:

- easy to expose (deploy) the current application to the production-ready environment
- built-in trajectory, forms integration, LLM inference support, ...
- easy to extend and debug


<CodeGroup>

	{/* <!-- embedme python/examples/serve/agent_stack.py --> */}

	```py Python [expandable]
	from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
	from beeai_framework.adapters.agentstack.serve.server import AgentStackMemoryManager, AgentStackServer
	from beeai_framework.agents.requirement import RequirementAgent
	from beeai_framework.backend import ChatModelParameters
	from beeai_framework.memory import UnconstrainedMemory
	from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
	from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
	from beeai_framework.tools.weather import OpenMeteoTool
	
	try:
	    from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail
	except ModuleNotFoundError as e:
	    raise ModuleNotFoundError(
	        "Optional module [agentstack] not found.\nRun 'pip install \"beeai-framework[agentstack]\"' to install."
	    ) from e
	
	
	def main() -> None:
	    llm = AgentStackChatModel(
	        preferred_models=["openai:gpt-4o", "ollama:llama3.1:8b"],
	        parameters=ChatModelParameters(stream=True),
	    )
	    agent = RequirementAgent(
	        llm=llm,
	        tools=[DuckDuckGoSearchTool(), OpenMeteoTool()],
	        memory=UnconstrainedMemory(),
	        middlewares=[GlobalTrajectoryMiddleware()],
	    )
	
	    # Runs HTTP server that registers to Agent Stack
	    server = AgentStackServer(memory_manager=AgentStackMemoryManager())
	    server.register(
	        agent,
	        name="Framework chat agent",  # (optional)
	        description="Simple chat agent",  # (optional)
	        detail=AgentDetail(interaction_mode="multi-turn"),  # default is multi-turn (optional)
	    )
	    server.serve()
	
	
	if __name__ == "__main__":
	    main()
	
	```

	```ts TypeScript [expandable]
	// COMING SOON
	```
</CodeGroup>


<Note>
	Agent Stack supports only one entry per server. To register more, you need to spawn more servers.
</Note>

<Tip>
	You can use the platform to do LLM inference by using the `AgentStackChatModel(...)` class.
</Tip>

### Configuration

**Server**

The server's behavior can be influenced via attributes listed in `AgentStackServerConfig` class (host, port, self registration, ...).
Internally the server preserves every conversation, the custom strategy can be used by implementing the base `MemoryManager` class.

```py Python [expandable]
from beeai_framework.adapters.agentstack.serve.server import AgentStackServer, AgentStackServerConfig
from beeai_framework.serve.utils import LRUMemoryManager

memory_manager = LRUMemoryManager(maxsize=64) # keeps max 64 conversations
config = AgentStackServerConfig(host="127.0.0.1", port=9999, run_limit=3600, limit_concurrency=10)
server = AgentStackServer(config=config, memory_manager=memory_manager)
```

**Agent**

The agent’s meta information is inferred from its metadata (the `agent.meta` property).
However, this information can be overridden during agent registration. See the following example.

```py Python [expandable]
from beeai_framework.adapters.agentstack.serve.server import AgentStackServer
from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail

server = AgentStackServer()
agent = RequirementAgent(llm=AgentStackChatModel())
server.register(agent,
	name="SmartAgent",
	description="Knows everything!",
	url="https://example.com",
	version="1.0.0",
	default_input_modes=["text", "text/plain"],
	default_output_modes=["text"],
	detail=AgentDetail(interaction_mode="multi-turn", user_greeting="What can I do for you?")
)
```

### Customization

The Agent Stack has a concept of [extensions](https://docs.beeai.dev/concepts/extensions) that enable access to external services and UI components via dependency injection.

The framework internally uses the following extensions:

- **Form Extension:** for displaying prompts and other checks (for instance when using `AskPermissionRequirement`)..
- **Trajectory Extension:** for showing agent's intermediate steps throughout the execution

<Note>The current entries are listed in the `BaseAgentStackExtensions` class.</Note>


**Custom Extensions**

The following implementation demonstrates an agent that conducts an internet search and provides an answer to the given question, with inline citations.
It does so by leveraging the [Citation Extension](https://docs.beeai.dev/build-agents/citations), which is managed by the  `PlatformCitationMiddleware` class.

<Note>Learn more about the [Middleware](/modules/middleware) concept.</Note>

<CodeGroup>

	{/* <!-- embedme python/examples/serve/agent_stack_custom.py --> */}

	```py Python [expandable]
	# Copyright 2025 © BeeAI a Series of LF Projects, LLC
	# SPDX-License-Identifier: Apache-2.0
	import re
	import sys
	import traceback
	from typing import Annotated
	
	from beeai_framework.adapters.agentstack.backend.chat import AgentStackChatModel
	from beeai_framework.adapters.agentstack.context import AgentStackContext
	from beeai_framework.adapters.agentstack.serve.server import AgentStackMemoryManager, AgentStackServer
	from beeai_framework.adapters.agentstack.serve.types import BaseAgentStackExtensions
	from beeai_framework.agents.requirement import RequirementAgent
	from beeai_framework.agents.requirement.events import RequirementAgentSuccessEvent
	from beeai_framework.agents.requirement.requirements.conditional import ConditionalRequirement
	from beeai_framework.backend import AssistantMessage
	from beeai_framework.context import RunContext, RunMiddlewareProtocol
	from beeai_framework.emitter import EmitterOptions, EventMeta
	from beeai_framework.errors import FrameworkError
	from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
	from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
	from beeai_framework.tools.search.wikipedia import WikipediaTool
	from beeai_framework.tools.think import ThinkTool
	
	try:
	    from agentstack_sdk.a2a.extensions import Citation, CitationExtensionServer, CitationExtensionSpec
	    from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail
	    from agentstack_sdk.a2a.types import AgentMessage
	except ModuleNotFoundError as e:
	    raise ModuleNotFoundError(
	        "Optional module [agentstack] not found.\nRun 'pip install \"beeai-framework[agentstack]\"' to install."
	    ) from e
	
	
	class CitationMiddleware(RunMiddlewareProtocol):
	    def __init__(self) -> None:
	        self._context: AgentStackContext | None = None
	
	    def bind(self, ctx: RunContext) -> None:
	        self._context = AgentStackContext.get()
	        # add emitter with the highest priority to ensure citations are sent before any other event handling
	        ctx.emitter.on("success", self._handle_success, options=EmitterOptions(priority=10, is_blocking=True))
	
	    async def _handle_success(self, data: RequirementAgentSuccessEvent, meta: EventMeta) -> None:
	        assert self._context is not None
	        citation_ext = self._context.extensions.get("citation")
	
	        # check it is the final step
	        if data.state.answer is not None:
	            citations, clean_text = extract_citations(data.state.answer.text)
	
	            if citations:
	                await self._context.context.yield_async(
	                    AgentMessage(metadata=citation_ext.citation_metadata(citations=citations))  # type: ignore[attr-defined]
	                )
	                # replace an assistant message with an updated text without citation links
	                data.state.answer = AssistantMessage(content=clean_text)
	
	
	# define custom extensions
	class CustomExtensions(BaseAgentStackExtensions):
	    citation: Annotated[CitationExtensionServer, CitationExtensionSpec()]
	
	
	def main() -> None:
	    agent = RequirementAgent(
	        llm=AgentStackChatModel(preferred_models=["openai/gpt-4o"]),
	        tools=[WikipediaTool(), ThinkTool(), DuckDuckGoSearchTool()],
	        instructions=(
	            "You are an AI assistant focused on retrieving information from online sources. "
	            "Mandatory Search: Always search for the topic on Wikipedia and always search for related current news. "
	            "Mandatory Output Structure: Return the result in two separate sections with headings: "
	            " 1. Basic Information (primarily utilizing data from Wikipedia, if relevant). "
	            " 2. News (primarily utilizing current news results). "
	            "Mandatory Citation: Always include a source link for all given information, especially news."
	        ),
	        requirements=[
	            ConditionalRequirement(ThinkTool, force_at_step=1, consecutive_allowed=False),
	            ConditionalRequirement(WikipediaTool, min_invocations=1),
	            ConditionalRequirement(DuckDuckGoSearchTool, min_invocations=1),
	        ],
	        description="Search for information based on a given phrase.",
	        middlewares=[
	            GlobalTrajectoryMiddleware(),
	            CitationMiddleware(),
	        ],  # add platform middleware to get citations from the platform
	    )
	
	    # Runs HTTP server that registers to Agent Stack
	    server = AgentStackServer(memory_manager=AgentStackMemoryManager())  # use platform memory
	    server.register(
	        agent,
	        name="Information retrieval",
	        detail=AgentDetail(interaction_mode="single-turn", user_greeting="What can I search for you?"),
	        extensions=CustomExtensions,
	    )
	    server.serve()
	
	
	# function to extract citations from text and return clean text without citation links
	def extract_citations(text: str) -> tuple[list[Citation], str]:
	    citations, offset = [], 0
	    pattern = r"\[([^\]]+)\]\(([^)]+)\)"
	
	    for match in re.finditer(pattern, text):
	        content, url = match.groups()
	        start = match.start() - offset
	
	        citations.append(
	            Citation(
	                url=url,
	                title=url.split("/")[-1].replace("-", " ").title() or content[:50],
	                description=content[:100] + ("..." if len(content) > 100 else ""),
	                start_index=start,
	                end_index=start + len(content),
	            )
	        )
	        offset += len(match.group(0)) - len(content)
	
	    return citations, re.sub(pattern, r"\1", text)
	
	
	if __name__ == "__main__":
	    try:
	        main()
	    except FrameworkError as e:
	        traceback.print_exc()
	        sys.exit(e.explain())
	
	```

	```ts TypeScript [expandable]
	// COMING SOON
	```

</CodeGroup>

<Tip>The `AgentStackContext.get()` can be called from anywhere if the code is running inside the given context.</Tip>
